package com.aaron.remoting.transport.netty.client;

import com.aaron.enums.CompressTypeEnum;
import com.aaron.enums.SerializationTypeEnum;
import com.aaron.factory.SingletonFactory;
import com.aaron.remoting.constants.RpcConstants;
import com.aaron.remoting.dto.RpcMessage;
import com.aaron.remoting.dto.RpcResponse;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

/**
 * 入站handler，客户端的入站操作就是 将字节流转化为 可读对象
 * 如果继承自 SimpleChannelInboundHandler 的话就不要考虑 ByteBuf 的释放 ，
 * 内部的channelRead 方法会替你释放 ByteBuf ，避免可能导致的内存泄露问题。
 */
@Slf4j
public class NettyRpcClientHandler extends ChannelInboundHandlerAdapter {
    // 未处理的rpcRequest，服务器没有处理的
    private final UnprocessedRequests unprocessedRequests;
    private final NettyRpcClient nettyRpcClient;

    public NettyRpcClientHandler() {
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
        this.nettyRpcClient = SingletonFactory.getInstance(NettyRpcClient.class);
    }

    // 读取从服务端返回的消息
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            log.info("client receive msg: [{}]", msg);
            if (msg instanceof RpcMessage) {
                RpcMessage tmp = (RpcMessage) msg;
                byte messageType = tmp.getMessageType();
                // 如果是建立连接，服务端回复的是pong
                if (messageType == RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
                    log.info("heart [{}]", tmp.getData());
                } else if (messageType == RpcConstants.RESPONSE_TYPE) { // 回复的是2，表示响应对象
                    RpcResponse<Object> rpcResponse = (RpcResponse<Object>) tmp.getData();
                    // unprocessedRequests中存放未处理请求,complete方法表示完成rpcResponse
                    // 那么就可以通过CompletableFuture.get()去拿了
                    unprocessedRequests.complete(rpcResponse);
                }
            }
        } finally {
            // 从InBound里读取的ByteBuf要手动释放，还有自己创建的ByteBuf要自己负责释放。
            ReferenceCountUtil.release(msg);
        }
    }

    // netty心跳机制。保证客户端和服务端的连接不被断掉，避免重连。定时发送简单的通讯包，如果在指定时间内未接收到对方响应，则判定对方已经宕掉。
    // 根据NettyRpcClient里面的配置是每5秒去判断是否执行了channelRead方法，若没被调用则调用本方法
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) { // 写超时事件
                log.info("write idle happen [{}]", ctx.channel().remoteAddress());
                // 获取管道
                Channel channel = nettyRpcClient.getChannel((InetSocketAddress) ctx.channel().remoteAddress());
                // 设置RpcMessage，发送ping消息
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                rpcMessage.setMessageType(RpcConstants.HEARTBEAT_REQUEST_TYPE);
                rpcMessage.setData(RpcConstants.PING);
                // 发送rpcMessage信息到远端，如果没有回应就关闭连接
                channel.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } else {
            // 传递给下一个处理程序
            super.userEventTriggered(ctx, evt);
        }
    }

    // 在处理客户端消息时发生异常时调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("client catch exception：", cause);
        cause.printStackTrace();
        ctx.close();
    }

}

